package redisson.rest.queue;

import java.util.ArrayList;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.redisson.api.RLock;
import org.redisson.api.RQueue;
import org.redisson.api.RedissonClient;
import org.redisson.codec.Kryo5Codec;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;
import redisson.rest.LocalCachedMapController;

@Slf4j
@Component
public class QueueConsumer implements SmartInitializingSingleton {

	@Autowired
	RedissonClient redissonClient;

	@Override
	public void afterSingletonsInstantiated() {
		////////////////////// 必须先配置监听//////////////////////
		log.info("### 监听redisson消息,启动成功,,, ");

		ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5,
				new CustomizableThreadFactory("demo-worker-"));
		// 入库
		scheduler.scheduleWithFixedDelay(this::flush, 10, 30, TimeUnit.SECONDS);
		log.info("### 模式订阅已建立");
	}

	private void flush() {
		RLock lock = redissonClient.getLock("monitor:rpa:lock");
		RQueue<Object> queue = redissonClient.getQueue("test:queue:axxc");
		ArrayList<Object> arrayList = new ArrayList<>();
		try {
			boolean ok = lock.tryLock(3, 30, TimeUnit.SECONDS);
			if (ok) {
				int size = queue.size();
				for (int i = 0; i < size; i++) {
					Object poll = queue.poll();
					if (Objects.isNull(poll)) {
						break;
					}
					arrayList.add(poll);
				}
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			if (lock.isHeldByCurrentThread()) {
				lock.unlock();
			}
		}
		if (!arrayList.isEmpty()) {
			log.info("### 此次获取了{}条消息!!!", arrayList.size());
		}

	}

}
